package com.heima.article.listener;

import com.alibaba.fastjson.JSON;


import java.time.Duration;

//@EnableBinding(IHotArticleProcessor.class)
/*
public class HotArticleListener {

    @StreamListener(value = "hot_article_score_topic_168")
    @SendTo(value = "hot_article_result_topic_168")
    public KStream<String, String> process(KStream<String, String> input) {
        // value 接收数据为 UpdateArticleMessage
        // 数据格式为  {"articleId":1483239346046881794,"type":0,"add":1}
        KStream<String, String> map = input.map(new KeyValueMapper<String, String, KeyValue<String, String>>() {
            @Override
            public KeyValue<String, String> apply(String key, String value) {
                // 解析json数据,转换成 UpdateArticleMessage
                UpdateArticleMessage updateArticleMessage = JSON.parseObject(value, UpdateArticleMessage.class);
                String articleId = updateArticleMessage.getArticleId().toString();
                // 构建KeyValue 使用文章id作为key,value还是原来的value
                return new KeyValue<>(articleId, value);
            }
        });
        // 需要根据文章id来进行分组
        KGroupedStream<String, String> groupByKey = map.groupByKey();
        // 需要统计每篇文章在10秒内的汇总数据
        TimeWindowedKStream<String, String> windowedBy = groupByKey.windowedBy(TimeWindows.of(Duration.ofSeconds(10)));

        // 结果数据为  ArticleStreamMessage
        // 前端发送消息次数为 5 + 3 +2+1= 11
        // 相当于 {"articleId":1483239346046881794,"type":0,"add":1} 5条消息
        // 相当于 {"articleId":1483239346046881794,"type":1,"add":1} 3条消息
        // 相当于 {"articleId":1483239346046881794,"type":2,"add":1} 2条消息
        // 相当于 {"articleId":1483239346046881794,"type":3,"add":1} 1条消息
        // 数据格式为  {"articleId":1483239346046881794,"view":5,"like":3,"comment":2,"collect":1}
        // 接收消息,保存到中间表(topic),将本次接收到的消息数据汇总到中间结果中进行累加
        // 累加完成后还保存回中间表
        Initializer<String> init = new Initializer<String>() {
            @Override
            public String apply() {
                // 聚合的中间结果第一次进来是空
                return null;
            }
        };

        Aggregator<String, String, String> aggregator = new Aggregator<String, String, String>() {
            @Override
            public String apply(String key, String value, String aggregate) {
                // 中间结果累加处理的逻辑
                // key 指的是上面的KeyValue 中的key,即文章id
                // value 指的是上面的KeyValue 中的 value,即 {"articleId":1483239346046881794,"type":0,"add":1}
                // aggregate 指的是上一次聚合处理的结果,第一次为空
                ArticleStreamMessage result = null;
                if (StringUtils.isEmpty(aggregate)) {
                    result = new ArticleStreamMessage();
                    result.setArticleId(Long.parseLong(key));
                    result.setView(0);
                    result.setLike(0);
                    result.setComment(0);
                    result.setCollect(0);
                } else {
                    result = JSON.parseObject(aggregate, ArticleStreamMessage.class);
                }
                // 解析本次接收到的消息
                UpdateArticleMessage updateArticleMessage = JSON.parseObject(value, UpdateArticleMessage.class);
                // 根据本次消息的操作类型进行数据的累加  操作类型 0 阅读 1 点赞 2 评论 3 收藏
                switch (updateArticleMessage.getType()) {
                    case 0:
                        result.setView(result.getView() + updateArticleMessage.getAdd());
                        break;
                    case 1:
                        result.setLike(result.getLike() + updateArticleMessage.getAdd());
                        break;
                    case 2:
                        result.setComment(result.getComment() + updateArticleMessage.getAdd());
                        break;
                    case 3:
                        result.setCollect(result.getCollect() + updateArticleMessage.getAdd());
                        break;
                }
                String json = JSON.toJSONString(result);
                return json;
            }
        };
        KTable<Windowed<String>, String> aggregate = windowedBy.aggregate(init, aggregator);
        // 将结果转换为字符串
        KStream<String, String> stream = aggregate.toStream().map(new KeyValueMapper<Windowed<String>, String, KeyValue<String, String>>() {
            @Override
            public KeyValue<String, String> apply(Windowed<String> key, String value) {
                return new KeyValue<>(key.key(), value);
            }
        });
        return stream;
    }
}
*/
